package cn.yi.boot.mq.rocketmq.manager;

import cn.yi.boot.mq.rocketmq.message.AbstractMessage;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;
import java.util.Optional;

@Slf4j
@Setter
public class DefaultMessageProducerManager {


    private String producerGroup;
    private String namesrvAddr;
    private int maxMessageSize;

    private DefaultMQProducer producer;

    public void init() throws MQClientException {
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(namesrvAddr);
        producer.setMaxMessageSize(maxMessageSize);
        producer.start();
    }

    public void distory() {
        Optional<DefaultMQProducer> optional = Optional.of(producer);
        if (Objects.nonNull(optional.get())) {
            producer.shutdown();
        }
    }

    /**
     * 发送消息方法
     *
     * @param message
     * @param <T>
     * @return
     */
    public <T> SendResult sendMessage(AbstractMessage<T> message) {
        try {
            return producer.send(message.getMessage());
        } catch (Exception e) {
            log.error("send mq message error", e);
            throw new RuntimeException("send mq message error");
        }
    }

    /**
     * 发送消息方法
     *
     * @param message
     * @param timeout
     * @param <T>
     * @return
     */
    public <T> SendResult sendMessage(AbstractMessage<T> message, long timeout) {
        try {
            return producer.send(message.getMessage(), timeout);
        } catch (Exception e) {
            log.error("send mq message error", e);
            throw new RuntimeException("send mq message error");
        }
    }
}
